package rabbit.topic;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;

/**
 * @Author: admin
 * @Description:
 * @Date: 2018/11/7 16:18
 */

public class TestRpcClient {

    ConnectionFactory connectionFactory;
    Connection connection;
    Channel channel;

    public TestRpcClient() throws IOException, TimeoutException {
        connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.10.66");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("root");
        connectionFactory.setPassword("root");
        connectionFactory.setVirtualHost("OuterHost");
        connection = connectionFactory.newConnection();
         channel = connection.createChannel();
    }


    public String call(String message) throws IOException, InterruptedException {
        final String corrId = UUID.randomUUID().toString();
        String replyQueueName =channel.queueDeclare().getQueue();
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().correlationId(corrId).replyTo(replyQueueName).build();
        channel.basicPublish("", "testrpcqueue", props, message.getBytes("UTF-8"));
        final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);
        String ctag = channel.basicConsume(replyQueueName, true, new DeliverCallback() {
            @Override
            public void handle(String s, Delivery delivery) throws IOException {
                if (delivery.getProperties().getCorrelationId().equals(corrId)) {
                    response.offer(new String(delivery.getBody(), "UTF-8"));
                }
            }
        }, new CancelCallback() {
            @Override
            public void handle(String s) throws IOException {

            }
        });
        String result=response.take();
        channel.basicCancel(ctag);
        return result;
    }

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        TestRpcClient client=new TestRpcClient();
        System.out.println( client.call("aaa"));
    }
}
